package com.ot.flink.cdc.function.source;

import com.ot.flink.cdc.config.CdcJdbcConfig;
import com.ot.flink.cdc.function.deserializer.CustomDebeziumDeserializationSchema;
import com.ot.flink.cdc.util.CdcStringUtil;
import com.ot.flink.cdc.util.SpringUtil;
import com.ververica.cdc.connectors.mysql.source.MySqlSource;
import com.ververica.cdc.connectors.mysql.source.MySqlSourceBuilder;
import com.ververica.cdc.connectors.mysql.table.StartupOptions;
import com.ververica.cdc.connectors.postgres.PostgreSQLSource;
import com.ververica.cdc.debezium.JsonDebeziumDeserializationSchema;
import org.apache.commons.lang3.StringUtils;
import org.apache.flink.streaming.api.functions.source.SourceFunction;

import java.util.List;
import java.util.Properties;
import java.util.UUID;
import java.util.stream.Collectors;


public class CdcSourceFactory {


    public static MySqlSource<String> createMysqlSource() {
        CdcJdbcConfig jdbcConfig = SpringUtil.getBean(CdcJdbcConfig.class);
        String jdbcUrl = jdbcConfig.getUrl();
        List<String> listTableFullName = jdbcConfig.getTables();
        List<String> listDb = listTableFullName.stream().map(str -> {
            return StringUtils.split(str, ".")[0];
        }).distinct().collect(Collectors.toList());
        // 使用MySQLSource创建数据源
        Properties properties = new Properties();
        // 日期格式化
        properties.setProperty("converters", "dateConverters");
        properties.setProperty("dateConverters.type", "com.ot.flink.cdc.function.convert.DebeziumConverter");
        properties.setProperty("dateConverters.format.date", "yyyy-MM-dd");
        properties.setProperty("dateConverters.format.time", "HH:mm:ss");
        properties.setProperty("dateConverters.format.datetime", "yyyy-MM-dd HH:mm:ss");
        properties.setProperty("dateConverters.format.timestamp", "yyyy-MM-dd HH:mm:ss");
        properties.setProperty("dateConverters.format.timestamp.zone", jdbcConfig.getZone());
        // initial(默认)	连接器执行数据库的初始一致性快照，快照完成后，连接器开始为后续数据库更改流式传输事件记录。
        //always	连接器在启动时始终执行一致性快照。快照完成后，连接器继续执行流式传输更改。
        //initial_only	连接器只执行数据库的初始一致性快照，不允许捕获任何后续更改的事件。
        //never	不执行初始一致性快照，但是会同步后续数据库的更改记录
        properties.setProperty("snapshot.mode", "never");
        properties.setProperty("debezium.slot.name", "pg_flink_cdc");
        properties.setProperty("debezium.slot.drop.on.stop", "true");
        properties.setProperty("include.schema.changes", "true");
        MySqlSourceBuilder<String> builder = MySqlSource.<String>builder()
                .hostname(StringUtils.split(jdbcUrl, ":")[0])
                .port(Integer.parseInt(StringUtils.split(jdbcUrl, ":")[1]))
                .username(jdbcConfig.getUsername())
                .password(jdbcConfig.getPassword())
                .databaseList(listDb.stream().toArray(String[]::new))
                .tableList(listTableFullName.stream().toArray(String[]::new))
                .debeziumProperties(properties)
                .deserializer(new JsonDebeziumDeserializationSchema());

        return builder.build();
    }

    /**
     * @Description: postgres
     * @author: admin
     * @date: 2024/4/19
     */
    public static SourceFunction<String> createPostgresSource() {
        CdcJdbcConfig jdbcConfig = SpringUtil.getBean(CdcJdbcConfig.class);
        Properties properties = new Properties();
        // 日期格式化
        properties.setProperty("converters", "dateConverters");
        properties.setProperty("dateConverters.type", "com.ot.flink.cdc.function.convert.DebeziumConverter");
        properties.setProperty("dateConverters.format.date", "yyyy-MM-dd");
        properties.setProperty("dateConverters.format.time", "HH:mm:ss");
        properties.setProperty("dateConverters.format.datetime", "yyyy-MM-dd HH:mm:ss");
        properties.setProperty("dateConverters.format.timestamp", "yyyy-MM-dd HH:mm:ss");
        properties.setProperty("dateConverters.format.timestamp.zone", jdbcConfig.getZone());
        /*properties.setProperty("database.history.instance.name", UUID.randomUUID().toString());
        properties.setProperty("database.history.skip.unparseable.ddl", "true");
        properties.setProperty("database.history.refer.ddl", "true");*/
        // initial(默认)	连接器执行数据库的初始一致性快照，快照完成后，连接器开始为后续数据库更改流式传输事件记录。
        //always	连接器在启动时始终执行一致性快照。快照完成后，连接器继续执行流式传输更改。
        //initial_only	连接器只执行数据库的初始一致性快照，不允许捕获任何后续更改的事件。
        //never	不执行初始一致性快照，但是会同步后续数据库的更改记录
        properties.setProperty("snapshot.mode", "never");
        properties.setProperty("debezium.slot.name", "pg_flink_cdc");
        properties.setProperty("debezium.slot.drop.on.stop", "true");
        properties.setProperty("include.schema.changes", "true");
        //使用连接器配置属性启用定期心跳记录生成
        //properties.setProperty("heartbeat.interval.ms", String.valueOf(DEFAULT_HEARTBEAT_MS));
        String jdbcUrl = jdbcConfig.getUrl();
        List<String> listTableFullName = jdbcConfig.getTables();
        List<String> listDb = listTableFullName.stream().map(str -> {
            return StringUtils.split(str, ".")[0];
        }).distinct().collect(Collectors.toList());

        List<String> listSchema = listTableFullName.stream().map(str -> {
            return StringUtils.split(str, ".")[1];
        }).distinct().collect(Collectors.toList());

        List<String> listTable = listTableFullName.stream().map(str -> {
            String[] split = StringUtils.split(str, ".");
            return split[1] + "." + split[2];
        }).distinct().collect(Collectors.toList());


        PostgreSQLSource.Builder<String> builder = PostgreSQLSource.<String>builder()
                .hostname(StringUtils.split(jdbcUrl, ":")[0])
                .port(Integer.parseInt(StringUtils.split(jdbcUrl, ":")[1]))
                .username(jdbcConfig.getUsername())
                .password(jdbcConfig.getPassword())
                .database(listDb.get(0))
                .schemaList(listSchema.stream().toArray(String[]::new))
                .tableList(listTable.stream().toArray(String[]::new))
                .slotName("pg_flink_cdc")
                //标准逻辑解码输出插件
                .decodingPluginName("pgoutput")
                .debeziumProperties(properties)
                .deserializer(new CustomDebeziumDeserializationSchema());

        return builder.build();
    }

}
